﻿// MIT License
// 开源地址：https://gitee.com/co1024/AbcMvc
// Copyright (c) 2021-2023 1024
// Abc.Mvc=Furion+EF+SqlSugar+Pear layui admin.

using Furion.EventBus;

namespace Abc.Core.EventBus;
public class EventBusHelper
{
    private static EventBusHelper instance = null;
    private static readonly object padlock = new object();

    private EventBusHelper()
    {
    }

    public static EventBusHelper Instance
    {
        get
        {
            if (instance == null)
            {
                lock (padlock)
                {
                    if (instance == null)
                    {
                        instance = new EventBusHelper();
                    }
                }
            }
            return instance;
        }
    }


    /// <summary>
    /// 获取事件ID，按Furion规则
    /// </summary>
    /// <param name="em"></param>
    /// <returns></returns>
    public string GetEventId(EventSubscribeType em)
    {
        var enumType = em.GetType();
        return $"{enumType.Assembly.GetName().Name};{enumType.FullName}.{em}";
    }



    /// <summary>
    /// 待处理的事件订阅数据，放内存，防止高频率读取数据库
    /// </summary>
    public ConcurrentQueue<SysEventSourceStorer> PendingEventSubscribeQueues = new ConcurrentQueue<SysEventSourceStorer>();

    /// <summary>
    /// 正在处理的事件订阅数据
    /// </summary>
    public ConcurrentDictionary<long, ProcessingEventSourceStorerDto> ProcessingEventSubscribes = new ConcurrentDictionary<long, ProcessingEventSourceStorerDto>();

    /// <summary>
    /// 设置正在处理的时间订阅数据 完成
    /// </summary>
    /// <param name="eventSourceStorer"></param>
    public void SetEventSubscribeCompleted(SysEventSourceStorer eventSourceStorer)
    {
        //完成时间
        eventSourceStorer.CompletedTime = DateTimeOffset.Now;
        //更新数据库处理状态
        if (eventSourceStorer.State == EventSourceStorerState.未处理)
        {
            eventSourceStorer.State = EventSourceStorerState.已处理;
        }

        SqlSugarHelper.Db.CopyNew().Updateable(eventSourceStorer)
           .UpdateColumns(o => new { o.State, o.CompletedTime, o.Error })
           .ExecuteCommand();

        //设置过期时间，防止重复处理
        if (ProcessingEventSubscribes.ContainsKey(eventSourceStorer.Id))
        {
            ProcessingEventSubscribes[eventSourceStorer.Id].ExpirationTime = DateTimeOffset.Now.AddSeconds(5);
        }
    }



    /// <summary>
    /// 读取一条数据,直到读取到数据才返回
    /// </summary>
    /// <returns></returns>
    public async Task<ChannelEventSource> ReadAsync(CancellationToken cancellationToken)
    {

        //判断缓存队列是否有数据，有就从队列取出，直接返回
        while (true)
        {
            try
            {
                var options = AppEx.GetConfig<EventBusOptions>();
                if (options == null) options = new EventBusOptions();//默认选项配置
                //如果事件总线没有启动
                if (options.IsEnableEventBus == false)
                {
                    //等3秒钟在执行
                    await Task.Delay(3 * 1000);
                    continue;
                }

                //判断正在处理的数据量是否超过最大同时处理的线程数
                if (EventBusHelper.Instance.ProcessingEventSubscribes.Count > options.ProcessingThreadMaxCount && options.ProcessingThreadMaxCount != 0)
                {
                    //等待xx毫秒，然后进行下次判断
                    await Task.Delay(50);
                    continue;
                }

                var state = PendingEventSubscribeQueues.TryDequeue(out SysEventSourceStorer data);
                if (state && data != null)
                {
                    //await Task.Delay(30);
                    return GetChannelEventSource(data);
                }
                else
                {
                    //等待50毫秒，直到取到数据才返回
                    await Task.Delay(30);
                }

            }
            catch (Exception ex)
            {
                //等待50毫秒，直到取到数据才返回
                await Task.Delay(50);
            }
        }
    }

    /// <summary>
    /// 获取  事件源
    /// </summary>
    /// <param name="eventSourceStorer"></param>
    /// <returns></returns>
    public ChannelEventSource GetChannelEventSource(SysEventSourceStorer eventSourceStorer)
    {
        //判断正在处理的 集合中是否存在，如果存在则返回null
        if (EventBusHelper.Instance.ProcessingEventSubscribes.ContainsKey(eventSourceStorer.Id)) return null;

        ////更新数据库中的状态
        //eventSourceStorer.State = EventSourceStorerState.处理中;
        //var resultcount = await SqlSugarDbContext.Db.Updateable(eventSourceStorer)
        //      .UpdateColumns(o => new { o.State, o.UpdateTime })
        //      .ExecuteCommandAsync();

        var options = AppEx.GetConfig<EventBusOptions>();
        if (options == null) options = new EventBusOptions();//默认选项配置

        //处理中的事件
        var processingEventSourceStorer = new ProcessingEventSourceStorerDto(eventSourceStorer);
        processingEventSourceStorer.MaxProcessingTime = DateTimeOffset.Now.AddSeconds(options.ExecTimeOutSeconds);//最大执行时间
        //处理中的事件  添加到处理中的集合中去
        EventBusHelper.Instance.ProcessingEventSubscribes.TryAdd(processingEventSourceStorer.EventSourceStorer.Id, processingEventSourceStorer);
        //返回 处理中的事件
        return new ChannelEventSource(processingEventSourceStorer.EventSourceStorer.EventId, processingEventSourceStorer);
    }


    /// <summary>
    /// 事件发布
    /// </summary>
    /// <param name="eventSubscribeType">事件类型</param>
    /// <param name="data">数据</param>
    public void Publisher(EventSubscribeType eventSubscribeType, object data)
    {
        var _logger = AppEx.GetSeriLogger($"EventBus\\事件发布\\{eventSubscribeType:G}");
        var str = $"事件类型：{eventSubscribeType:G}\r\n事件内容：{data.ToJson()}";
        Retry.Invoke(() =>
        {
            try
            {
                var eventSourceStorer = new SysEventSourceStorer();
                eventSourceStorer.EventId = GetEventId(eventSubscribeType);
                eventSourceStorer.Payload = data.ToJson();

                var result = SqlSugarHelper.Db.CopyNew().InsertRsId(eventSourceStorer);
                str += $"新增返回ID：{result}";
            }
            catch (Exception ex)
            {
                str += $"异常信息：{ex.ToStringEx()}";
                _logger.Error(str);
                throw;
            }
            finally
            {
                _logger.Debug(str);
            }
        }, 5, 1000, fallbackPolicy: (ex) =>
        {
            _logger.Error($"{str}\r\n异常：{ex.ToStringEx()}");
        });
    }

    /// <summary>
    /// 事件发布
    /// </summary>
    /// <param name="eventSubscribeType">事件类型</param>
    /// <param name="data">数据</param>
    public async Task PublisherAsync(EventSubscribeType eventSubscribeType, object data)
    {
        var _logger = AppEx.GetSeriLogger($"EventBus\\事件发布\\{eventSubscribeType:G}");
        var str = $"事件类型：{eventSubscribeType:G}\r\n事件内容：{data.ToJson()}";
        await Retry.InvokeAsync(async () =>
        {
            try
            {
                var eventSourceStorer = new SysEventSourceStorer();
                eventSourceStorer.EventId = GetEventId(eventSubscribeType); ;
                eventSourceStorer.Payload = data.ToJson();
                var result = await SqlSugarHelper.Db.CopyNew().InsertRsIdAsync(eventSourceStorer);
                str += $"新增返回ID：{result}";
            }
            catch (Exception ex)
            {
                str += $"异常信息：{ex.ToStringEx()}";
                _logger.Error(str);
                throw;
            }
            finally
            {
                _logger.Debug(str);
            }
        }, 5, 1000, fallbackPolicy: async (ex) =>
        {
            _logger.Error($"{str}\r\n异常：{ex.ToStringEx()}");
            await Task.CompletedTask;
        });
    }

}
